package ws

import (
	"encoding/json"
	"fmt"
	"gitee.com/scottq/go-framework/src/v1/libs/comag"
	"github.com/gorilla/websocket"
	"github.com/rfyiamcool/backoff"
	"net/http"
	"sync/atomic"
	"time"
)

var wsDialer = websocket.Dialer{
	ReadBufferSize:  1024,
	WriteBufferSize: 1024,
	Proxy:           http.ProxyFromEnvironment,
}

type WsConnector interface {
	SendJson(msg chan interface{})
	Send(msg []byte)
	Start(reader chan []byte) error
	Conn() *websocket.Conn
}

// ConnectorEvent 连接器事件
type ConnectorEvent string

const (
	ConnectNotify ConnectorEvent = "connect notify"
)

// wsConnector ws连接器
type wsConnector struct {
	comag.StopWorker

	wsURL       string
	conn        *websocket.Conn
	reader      chan []byte
	sendCh      chan []byte
	retry       int
	eventHandle func(ConnectorEvent)

	closed atomic.Bool
}

// NewFeatureWsConnector ws辅助连接器
func NewFeatureWsConnector(url string, retry int) *wsConnector {
	return &wsConnector{
		wsURL:  url,
		sendCh: make(chan []byte, 1),
		retry:  retry,
	}
}

func (connector *wsConnector) Conn() *websocket.Conn {
	return connector.conn
}

func (connector *wsConnector) Close() error {
	connector.closed.Swap(true)
	return connector.conn.Close()
}

func (connector *wsConnector) Start(reader chan []byte) error {
	err := connector.reConnect()
	if err != nil {
		logger.Error("ws reconnect fail! %s\n", connector.wsURL)
		return err
	}
	logger.Info("ws start connector %s", connector.wsURL)
	connector.reader = reader

	go connector.read()
	go connector.write()

	return nil
}

func (connector *wsConnector) SendJson(msg chan interface{}) {
	p, _ := json.Marshal(msg)
	connector.Send(p)
}

func (connector *wsConnector) Send(msg []byte) {
	connector.sendCh <- msg
}

func (connector *wsConnector) read() {
	defer func() {
		logger.Info("ws connector write stop %s", connector.wsURL)
	}()
	for {
		connector.conn.SetReadDeadline(time.Now().Add(pongWait))
		_, p, err := connector.conn.ReadMessage()
		if err != nil {
			logger.Info("ws reconnect be overdue! %s %s", err, connector.wsURL)
			if connector.closed.Load() {
				return
			}
			//断线重连
			err = connector.reConnect()
			if err != nil {
				logger.Error("ws reconnect fail! %s %s", err, connector.wsURL)
			}
			continue
		}
		if connector.reader != nil {
			connector.reader <- p
		}
	}
}

func (connector *wsConnector) write() {
	defer func() {
		logger.Info("ws connector write stop %s", connector.wsURL)
	}()
	for {
		select {
		case p, ok := <-connector.sendCh:
			if !ok {
				return
			}
			if connector.conn == nil {
				continue
			}
			if connector.closed.Load() {
				return
			}
			if err := connector.conn.WriteMessage(websocket.TextMessage, p); err != nil {
				logger.Error("write message err: %s %s", err, connector.wsURL)
			}
		}
	}
}

// connect 单次连接
func (connector *wsConnector) connect() (*websocket.Conn, error) {
	url := connector.wsURL
	c, _, err := wsDialer.Dial(url, nil)
	if err != nil {
		return nil, err
	}
	return c, nil
}

// reConnect 重试连接
func (connector *wsConnector) reConnect() error {
	//指数退避重连
	maxRecon := connector.retry
	b := backoff.NewBackOff(
		backoff.WithMinDelay(1*time.Second),
		backoff.WithMaxDelay(10*time.Second),
		backoff.WithFactor(2),
		backoff.WithJitterFlag(true),
	)
	var final error

	for index := 0; index <= maxRecon; index++ {
		c, err := connector.connect()
		if err != nil {
			logger.Error("ws connect fail which %d %s !", index, err)
			final = err
			time.Sleep(b.Duration())
			continue
		}
		c.SetReadLimit(maxMessageSize)
		c.SetReadDeadline(time.Now().Add(pongWait))
		c.SetPongHandler(func(string) error {
			connector.conn.SetReadDeadline(time.Now().Add(pongWait))
			return nil
		})

		connector.conn = c
		//事件通知
		if connector.eventHandle != nil {
			go connector.eventHandle(ConnectNotify)
		}
		logger.Info("ws connect success %s", connector.wsURL)
		return nil
	}
	return fmt.Errorf("ws connect fail %s", final)
}

func (connector *wsConnector) SetEventHandle(f func(event ConnectorEvent)) {
	connector.eventHandle = f
}
